package com.haozhen.streaming

/**
  * @author haozhen
  * @email haozh@ync1.com
  * @date 2021/1/16  20:11
  *
  *  不支持嵌套
  *  另存的文件 注意时间戳
  */
object FileDstream {

  def main(args: Array[String]): Unit = {
    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.dstream.DStream
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    Logger.getLogger("org").setLevel(Level.WARN)

    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val ssc = new StreamingContext(conf,Seconds(10))

    val lines: DStream[String] = ssc.textFileStream("data/log")

    //DStream 转换
    val words: DStream[String] = lines.flatMap(_.split("\\s+"))
    val result: DStream[(String, Int)] = words.map((_,1)).reduceByKey(_+_)

    //DStream 输出
    result.print(20)
    //启动作业
    ssc.start()
    ssc.awaitTermination()
  }
}
